Node / meshtastic / Meshtastic-Android / files / core / data / src / commonTest / kotlin / org / meshtastic / core / data / datasource / PoisonedPoolNodeFlowRecoveryTest.kt

Displaying Raw • Download

T8b949e/*

* Copyright (c) 2026 Meshtastic LLC

*

* This program is free software: you can redistribute it and/or modify

* it under the terms of the GNU General Public License as published by

* the Free Software Foundation, either version 3 of the License, or

* (at your option) any later version.

*

* This program is distributed in the hope that it will be useful,

* but WITHOUT ANY WARRANTY; without even the implied warranty of

* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

* GNU General Public License for more details.

*

* You should have received a copy of the GNU General Public License

* along with this program. If not, see <https://www.gnu.org/licenses/>.

*/

Tff7b72package T7ee787org.meshtastic.core.data.datasource

Tff7b72import T7ee787app.cash.turbine.test

Tff7b72import T7ee787kotlinx.coroutines.ExperimentalCoroutinesApi

Tff7b72import T7ee787kotlinx.coroutines.flow.Flow

Tff7b72import T7ee787kotlinx.coroutines.flow.MutableStateFlow

Tff7b72import T7ee787kotlinx.coroutines.flow.SharingStarted

Tff7b72import T7ee787kotlinx.coroutines.flow.StateFlow

Tff7b72import T7ee787kotlinx.coroutines.flow.emitAll

Tff7b72import T7ee787kotlinx.coroutines.flow.flatMapLatest

Tff7b72import T7ee787kotlinx.coroutines.flow.flow

Tff7b72import T7ee787kotlinx.coroutines.flow.map

Tff7b72import T7ee787kotlinx.coroutines.flow.stateIn

Tff7b72import T7ee787kotlinx.coroutines.test.runTest

Tff7b72import T7ee787org.meshtastic.core.database.DatabaseProvider

Tff7b72import T7ee787org.meshtastic.core.database.MeshtasticDatabase

Tff7b72import T7ee787org.meshtastic.core.database.getInMemoryDatabaseBuilder

Tff7b72import T7ee787kotlin.test.AfterTest

Tff7b72import T7ee787kotlin.test.Test

Tff7b72import T7ee787kotlin.test.assertEquals

Tff7b72import T7ee787kotlin.test.assertTrue

Tff7b72import T7ee787kotlin.time.Duration.Companion.minutes

T8b949e/**

* Regression for #6608. A Room permit leak surfaces as an endless acquire-timeout storm, and

* `DatabaseProvider.observeCurrentDb` recovery is budgeted — once the budget is spent, the DAO flow terminates. Before

* this hardening, a terminated upstream under `stateIn(..., SharingStarted.Eagerly, ...)` left the node list empty for

* the rest of the process. These tests poison the pool for more failures than any recovery budget allows and assert the

* node flows still reach a value.

*/

Tf0883e@OptInTb4b4b4(Te6edf3ExperimentalCoroutinesApiTff7b72::Te6edf3classTb4b4b4)

Tff7b72class T56d364PoisonedPoolNodeFlowRecoveryTest Tb4b4b4{

Tff7b72private Tff7b72val Te6edf3database Tff7b72= Te6edf3getInMemoryDatabaseBuilderTb4b4b4(Tb4b4b4)Tb4b4b4.Te6edf3buildTb4b4b4(Tb4b4b4)

Tf0883e@AfterTest Tff7b72fun Td2a8fftearDownTb4b4b4(Tb4b4b4) Tff7b72= Te6edf3databaseTb4b4b4.Te6edf3closeTb4b4b4(Tb4b4b4)

Tf0883e@Test

Tff7b72fun Td2a8ffnodeDbFlowRecoversAfterMoreFailuresThanTheRecoveryBudgetAllowsTb4b4b4(Tb4b4b4) Tff7b72= Te6edf3runTest Tb4b4b4{

Tff7b72val Te6edf3provider Tff7b72= Te6edf3PoisonedPoolProviderTb4b4b4(Te6edf3databaseTb4b4b4, Te6edf3failuresBeforeRecovery Tff7b72= Te6edf3POISONED_FAILURESTb4b4b4)

Tff7b72val Te6edf3dataSource Tff7b72= Te6edf3SwitchingNodeInfoReadDataSourceTb4b4b4(Te6edf3providerTb4b4b4)

Te6edf3dataSourceTb4b4b4.Te6edf3nodeDBbyNumFlowTb4b4b4(Tb4b4b4)Tb4b4b4.Te6edf3testTb4b4b4(Te6edf3timeout Tff7b72= Te6edf3EMISSION_TIMEOUTTb4b4b4) Tb4b4b4{

Te6edf3assertEqualsTb4b4b4(Te6edf3emptyMapTb4b4b4(Tb4b4b4)Tb4b4b4, Te6edf3awaitItemTb4b4b4(Tb4b4b4)Tb4b4b4)

Te6edf3cancelAndIgnoreRemainingEventsTb4b4b4(Tb4b4b4)

Tb4b4b4}

Te6edf3assertTrueTb4b4b4(Te6edf3providerTb4b4b4.Te6edf3observedQueries Tff7b72> T79c0ff1Tb4b4b4, Ta5d6ff"Ta5d6ffthe wedged flow must be restarted, not left terminalTa5d6ff"Tb4b4b4)

Tb4b4b4}

T8b949e/** Proves the fix at the layer that actually broke: an eagerly-shared StateFlow over the poisoned DAO flow. */

Tf0883e@Test

Tff7b72fun Td2a8ffeagerlySharedNodeStateFlowStillPublishesAfterAPoisonedPoolTb4b4b4(Tb4b4b4) Tff7b72= Te6edf3runTest Tb4b4b4{

Tff7b72val Te6edf3provider Tff7b72= Te6edf3PoisonedPoolProviderTb4b4b4(Te6edf3databaseTb4b4b4, Te6edf3failuresBeforeRecovery Tff7b72= Te6edf3POISONED_FAILURESTb4b4b4)

Tff7b72val Te6edf3dataSource Tff7b72= Te6edf3SwitchingNodeInfoReadDataSourceTb4b4b4(Te6edf3providerTb4b4b4)

Te6edf3dataSource

Tb4b4b4.Te6edf3nodeDBbyNumFlowTb4b4b4(Tb4b4b4)

Tb4b4b4.Te6edf3map Tb4b4b4{ Te6edf3map Tff7b72-Tff7b72> Te6edf3mapTb4b4b4.Te6edf3keysTb4b4b4.Te6edf3toSetTb4b4b4(Tb4b4b4) Tb4b4b4}

Tb4b4b4.Te6edf3stateInTb4b4b4(Te6edf3backgroundScopeTb4b4b4, Te6edf3SharingStartedTb4b4b4.Te6edf3EagerlyTb4b4b4, Tff7b72nullTb4b4b4)

Tb4b4b4.Te6edf3testTb4b4b4(Te6edf3timeout Tff7b72= Te6edf3EMISSION_TIMEOUTTb4b4b4) Tb4b4b4{

T8b949e// The seed null is the StateFlow's initial value; the set proves the shared upstream survived.

Te6edf3assertEqualsTb4b4b4(Te6edf3emptySetTb4b4b4(Tb4b4b4)Tb4b4b4, Te6edf3awaitItemTb4b4b4(Tb4b4b4) Tff7b72?: Te6edf3awaitItemTb4b4b4(Tb4b4b4)Tb4b4b4)

Te6edf3cancelAndIgnoreRemainingEventsTb4b4b4(Tb4b4b4)

Tb4b4b4}

Tb4b4b4}

Tf0883e@Test

Tff7b72fun Td2a8ffmyNodeInfoFlowRecoversAfterAPoisonedPoolTb4b4b4(Tb4b4b4) Tff7b72= Te6edf3runTest Tb4b4b4{

Tff7b72val Te6edf3provider Tff7b72= Te6edf3PoisonedPoolProviderTb4b4b4(Te6edf3databaseTb4b4b4, Te6edf3failuresBeforeRecovery Tff7b72= Te6edf3POISONED_FAILURESTb4b4b4)

Tff7b72val Te6edf3dataSource Tff7b72= Te6edf3SwitchingNodeInfoReadDataSourceTb4b4b4(Te6edf3providerTb4b4b4)

Te6edf3dataSourceTb4b4b4.Te6edf3myNodeInfoFlowTb4b4b4(Tb4b4b4)Tb4b4b4.Te6edf3testTb4b4b4(Te6edf3timeout Tff7b72= Te6edf3EMISSION_TIMEOUTTb4b4b4) Tb4b4b4{

Te6edf3assertEqualsTb4b4b4(Tff7b72nullTb4b4b4, Te6edf3awaitItemTb4b4b4(Tb4b4b4)Tb4b4b4)

Te6edf3cancelAndIgnoreRemainingEventsTb4b4b4(Tb4b4b4)

Tb4b4b4}

Tb4b4b4}

T8b949e/**

* Models a leaked-permit pool with an exhausted recovery budget: the first [failuresBeforeRecovery] collections

* throw Room's acquire-timeout message and no in-place pool replacement happens, exactly as when

* `DatabaseManager`'s Flow-recovery budget refuses another reopen.

*/

Tff7b72private Tff7b72class T56d364PoisonedPoolProviderTb4b4b4(Te6edf3databaseTb4b4b4: Te6edf3MeshtasticDatabaseTb4b4b4, Tff7b72private Tff7b72val Te6edf3failuresBeforeRecoveryTb4b4b4: Tffa657IntTb4b4b4) Tb4b4b4:

Te6edf3DatabaseProvider Tb4b4b4{

Tff7b72private Tff7b72val Te6edf3mutableCurrentDb Tff7b72= Te6edf3MutableStateFlowTb4b4b4(Te6edf3databaseTb4b4b4)

Tff7b72override Tff7b72val Te6edf3currentDbTb4b4b4: Te6edf3StateFlowTff7b72<Te6edf3MeshtasticDatabaseTff7b72> Tff7b72= Te6edf3mutableCurrentDb

Tff7b72var Te6edf3observedQueriesTb4b4b4: Tffa657Int Tff7b72= T79c0ff0

Te6edf3private Tff7b72set

Tff7b72override Tff7b72fun Tff7b72<Te6edf3TTff7b72> Td2a8ffobserveCurrentDbTb4b4b4(Te6edf3queryTb4b4b4: Tb4b4b4(Te6edf3MeshtasticDatabaseTb4b4b4) Tff7b72-Tff7b72> Te6edf3FlowTff7b72<Te6edf3TTff7b72>Tb4b4b4)Tb4b4b4: Te6edf3FlowTff7b72<Te6edf3TTff7b72> Tff7b72=

Te6edf3currentDbTb4b4b4.Te6edf3flatMapLatest Tb4b4b4{ Te6edf3database Tff7b72-Tff7b72>

Te6edf3flow Tb4b4b4{

Te6edf3observedQueries Tff7b72+Tff7b72= T79c0ff1

Tff7b72if Tb4b4b4(Te6edf3observedQueries Tff7b72<Tff7b72= Te6edf3failuresBeforeRecoveryTb4b4b4) Tb4b4b4{

Tff7b72throw Te6edf3IllegalStateExceptionTb4b4b4(

Ta5d6ff"Ta5d6ffError code: 5, message: Timed out attempting to acquire a reader connectionTa5d6ff"Tb4b4b4,

Tb4b4b4)

Tb4b4b4}

Te6edf3emitAllTb4b4b4(Te6edf3queryTb4b4b4(Te6edf3databaseTb4b4b4)Tb4b4b4)

Tb4b4b4}

Tb4b4b4}

Tff7b72override Tff7b72suspend Tff7b72fun Tff7b72<Te6edf3TTff7b72> Td2a8ffwithReadDbTb4b4b4(Te6edf3blockTb4b4b4: Te6edf3suspend Tb4b4b4(Te6edf3MeshtasticDatabaseTb4b4b4) Tff7b72-Tff7b72> Te6edf3TTb4b4b4)Tb4b4b4: Te6edf3T Tff7b72= Te6edf3blockTb4b4b4(Te6edf3currentDbTb4b4b4.Te6edf3valueTb4b4b4)

Tff7b72override Tff7b72suspend Tff7b72fun Tff7b72<Te6edf3TTff7b72> Td2a8ffwithDbTb4b4b4(Te6edf3blockTb4b4b4: Te6edf3suspend Tb4b4b4(Te6edf3MeshtasticDatabaseTb4b4b4) Tff7b72-Tff7b72> Te6edf3TTb4b4b4)Tb4b4b4: Te6edf3T? Tff7b72= Te6edf3blockTb4b4b4(Te6edf3currentDbTb4b4b4.Te6edf3valueTb4b4b4)

Tb4b4b4}

Tff7b72private Tff7b72companion Tff7b72object Tb4b4b4{

T8b949e/** Far more failures than any per-collector or per-window recovery budget allows. */

Tff7b72const Tff7b72val Te6edf3POISONED_FAILURES Tff7b72= T79c0ff2T79c0ff4

T8b949e// Turbine awaits run on the test scheduler's virtual clock, which the retry backoff also advances, so this

T8b949e// bound covers the accumulated capped backoff rather than any wall-clock expectation.

Tff7b72val Te6edf3EMISSION_TIMEOUT Tff7b72= T79c0ff3T79c0ff0.Te6edf3minutes

Tb4b4b4}

Tb4b4b4}